package com.bigdata.hdfs;

import com.bigdata.commom.PropConstant;
import com.common.utils.PropertiesUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.URI;
import java.text.MessageFormat;

/**
 * @author xin.ding
 * @date 2020/10/17 9:21
 */
public class HdfsOper {

    private static final Logger logger = LoggerFactory.getLogger(HdfsOper.class);

    private String hdfsUrl;
    private String hdfsUser;
    private String hdfsPrincipal;
    private String krb5Path;
    private String keytabPath;
    private boolean isKerberos;
    private FileSystem fileSystem;


    public HdfsOper() {
        buildFileSystem();
    }

    private void init() {
        hdfsUrl = PropertiesUtils.getString(PropConstant.HDFS_URL);
        hdfsUser = PropertiesUtils.getString(PropConstant.HDFS_USER);
        hdfsPrincipal = PropertiesUtils.getString(PropConstant.HDFS_PRINCIPAL);
        krb5Path = PropertiesUtils.getString(PropConstant.HDFS_KRB5_PATH);
        keytabPath = PropertiesUtils.getString(PropConstant.HDFS_KEYTAB);
        isKerberos = PropertiesUtils.getBoolean(PropConstant.HDFS_KERBEROS);
    }

    private FileSystem getFileSystem() {
        if (fileSystem == null) {
            buildFileSystem();
        }
        return fileSystem;
    }


    private void buildFileSystem() {
        init();
        if (isKerberos) {
            fileSystem = createKrbFileSystem(krb5Path, hdfsPrincipal, keytabPath);
        } else {
            fileSystem = createFileSystem(hdfsUrl, hdfsUser, null);
        }
    }


    /**
     * 初始化Kerberos环境
     *
     * @param krb5Path   krb5.conf路径
     * @param user       keytab principal 名称
     * @param keytabPath keytab文件路径
     */
    public FileSystem createKrbFileSystem(String krb5Path, String user, String keytabPath) {
        FileSystem fs = null;
        try {
            System.setProperty("java.security.krb5.conf", krb5Path);
            Configuration configuration = new Configuration();
            configuration.set("hadoop.security.authentication", "Kerberos");
            //设置使用hdfs分布式文件系统
            configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
            configuration.set("dfs.client.use.datanode.hostname", "true");

            UserGroupInformation.setConfiguration(configuration);
            UserGroupInformation.loginUserFromKeytab(user, keytabPath);
            logger.info("kerberos login info:{}", UserGroupInformation.getCurrentUser());
            fs = FileSystem.get(URI.create(hdfsUrl), configuration);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return fs;
    }


    /**
     * 获得fs对象
     *
     * @param con     hdfs连接地址，例如：http://hadoop11:8082
     * @param user    hdfs用户,例如
     * @param confNew Hdfs配置，默认配置可传入null
     * @return FileSystem
     */
    public FileSystem createFileSystem(String con, String user, Configuration confNew) {
        FileSystem fs = null;
        URI uri;
        try {
            uri = new URI(con);
            Configuration conf = new Configuration();
            conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
            conf.set("dfs.client.use.datanode.hostname", "true");
            if (confNew != null) {
                fs = FileSystem.get(uri, confNew, user);
            } else {
                fs = FileSystem.get(uri, conf, user);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return fs;
    }

    /**
     * 关闭fs
     */
    public void closeFS() {
        if (fileSystem != null) {
            try {
                fileSystem.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    public void list(String folder) {
        if (!folder.endsWith("/")) {
            folder += "/";
        }
        logger.info("==========ls {}===========", folder);
        FileStatus[] list;
        try {
            list = getFileSystem().listStatus(new Path(folder));
            for (FileStatus f : list) {
                System.out.println(folder + f.getPath().getName() + "\t" + f.getOwner());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    /**
     * io流方式上传文件至hdfs
     *
     * @param fs       hdfs文件系统
     * @param path     本地文件路径
     * @param hdfsPath hdfs文件路径
     */
    public static void uploadToHDFS(FileSystem fs, String path, String hdfsPath) {
        try (FileInputStream fis = new FileInputStream(path);
             FSDataOutputStream os = fs.create(new Path(hdfsPath));) {
            IOUtils.copyBytes(fis, os, 3000, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * io流方式从hdfs下载文件
     *
     * @param fs        HDFS文件系统
     * @param localPath 本地路径
     * @param hdfsPath  hdfs路径
     */
    public static void downloadFromHDFS(FileSystem fs, String localPath, String hdfsPath) {
        try (FSDataInputStream is = fs.open(new Path(hdfsPath));
             FileOutputStream fos = new FileOutputStream(localPath)) {
            IOUtils.copyBytes(is, fos, 2000, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 创建目录 支持递归创建
     *
     * @param path 路径地址
     * @return 创建是否成功
     */
    public boolean mkdir(String path) throws Exception {
        return getFileSystem().mkdirs(new Path(path));
    }

    /**
     * 查看文件内容
     *
     * @param path 路径地址
     * @return 返回文件内容字符串
     */
    public String cat(String path) throws Exception {
        return this.cat(path, "UTF-8");
    }

    /**
     * 查看文件内容
     *
     * @param path 路径地址
     * @return 返回文件内容字符串
     */
    public String cat(String path, String encode) throws Exception {
        FSDataInputStream inputStream = getFileSystem().open(new Path(path));
        return inputStreamToString(inputStream, encode);
    }


    /**
     * 创建文件并写入内容
     *
     * @param path    路径地址
     * @param context 文件内容
     */
    public void createAndWrite(String path, String context) throws Exception {
        FSDataOutputStream out = getFileSystem().create(new Path(path));
        out.write(context.getBytes());
        out.flush();
        out.close();
    }

    /**
     * 文件重命名
     *
     * @param oldPath 旧文件路径
     * @param newPath 新文件路径
     * @return 重命名是否成功
     */
    public boolean rename(String oldPath, String newPath) throws Exception {
        return getFileSystem().rename(new Path(oldPath), new Path(newPath));

    }

    /**
     * 上传文件至HDFS
     *
     * @param srcPath  本地文件路径，比如：D:/test.txt
     * @param hdfsPath HDFS的相对目录路径，比如：/testDir
     */
    public void uploadFileToHdfs(String srcPath, String hdfsPath) {
        this.uploadFileToHdfs(false, true, srcPath, hdfsPath);
    }


    /**
     * 上传文件至HDFS
     *
     * @param delSrc    是否删除本地文件
     * @param overwrite 是否覆盖HDFS上面的文件
     * @param srcFile   本地文件路径，比如：D:/test.txt
     * @param dstPath   HDFS的相对目录路径，比如：/testDir
     */
    public void uploadFileToHdfs(boolean delSrc, boolean overwrite, String srcFile, String dstPath) {
        try {
            //先判断文件夹是否存在
            boolean flag = mkdir(dstPath);
            if (flag) {
                // 源文件路径
                Path localSrcPath = new Path(srcFile);
                // 目标文件路径
                Path hdfsDstPath = new Path(generateHdfsPath(dstPath));
                getFileSystem().copyFromLocalFile(delSrc, overwrite, localSrcPath, hdfsDstPath);
            }

        } catch (Exception e) {
            logger.error(MessageFormat.format("上传文件至HDFS失败，srcFile:{0},dstPath:{1}", srcFile, dstPath), e);
            throw new RuntimeException(MessageFormat.format("上传文件至HDFS失败，srcFile:{0},dstPath:{1}", srcFile, dstPath), e);
        } finally {
            closeFS();
        }
    }

    /**
     * 将相对路径转化为HDFS文件路径
     *
     * @param dstPath 相对路径，比如：/data
     * @return java.lang.String
     */
    private String generateHdfsPath(String dstPath) {
        String hdfsPath = "";
        if (dstPath.startsWith("/")) {
            hdfsPath += dstPath;
        } else {
            hdfsPath = hdfsPath + "/" + dstPath;
        }

        return hdfsPath;
    }

    /**
     * 拷贝文件到HDFS
     *
     * @param localPath 本地文件路径
     * @param hdfsPath  存储到hdfs上的路径
     */
    public void copyFromLocalFile(String localPath, String hdfsPath) throws Exception {
        getFileSystem().copyFromLocalFile(new Path(localPath), new Path(hdfsPath));
    }


    /**
     * 从HDFS下载文件
     *
     * @param hdfsPath  文件在hdfs上的路径
     * @param localPath 存储到本地的路径
     */
    public void copyToLocalFile(String hdfsPath, String localPath) throws Exception {
        getFileSystem().copyToLocalFile(new Path(hdfsPath), new Path(localPath));
    }

    /**
     * 从HDFS下载文件至本地
     *
     * @param hdfsPath  HDFS的相对目录路径，比如：/testDir/a.txt
     * @param localPath 下载之后本地文件路径（如果本地文件目录不存在，则会自动创建），比如：D:/test.txt
     */
    public void downloadFileFromHdfs(String hdfsPath, String localPath) {
        // HDFS文件路径
        Path hdfsSrcPath = new Path(generateHdfsPath(hdfsPath));
        // 下载之后本地文件路径
        Path localDstPath = new Path(localPath);

        try {
            getFileSystem().copyToLocalFile(hdfsSrcPath, localDstPath);
        } catch (Exception e) {
            logger.error(MessageFormat.format("从HDFS下载文件至本地失败，hdfsPath:{0},localPath:{1}", hdfsPath, localPath), e);
            throw new RuntimeException(
                    MessageFormat.format("从HDFS下载文件至本地失败，hdfsPath:{0},localPath:{1}", hdfsPath, localPath));
        } finally {
            closeFS();
        }
    }


    /**
     * 查看文件块信息
     *
     * @param path 文件路径
     * @return 块信息数组
     */
    public BlockLocation[] getFileBlockLocations(String path) throws Exception {
        FileStatus fileStatus = fileSystem.getFileStatus(new Path(path));
        return fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    }

    /**
     * 删除文件
     *
     * @param path 文件路径
     * @return 删除是否成功
     */
    public boolean delete(String path) throws Exception {
        return getFileSystem().delete(new Path(path), true);
    }


    /**
     * 把输入流转换为指定字符
     *
     * @param inputStream 输入流
     * @param encode      指定编码类型
     */
    private static String inputStreamToString(InputStream inputStream, String encode) {
        try {
            if (encode == null || ("".equals(encode))) {
                encode = "utf-8";
            }
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
            StringBuilder builder = new StringBuilder();
            String str = "";
            while ((str = reader.readLine()) != null) {
                builder.append(str).append("\n");
            }
            return builder.toString();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 初始化HDFS Configuration
     *
     * @return configuration
     */
    public static Configuration initConfiguration(String confPath) {
        Configuration configuration = new Configuration();
        System.out.println(confPath + File.separator + "core-site.xml");
        configuration.addResource(new Path(confPath + File.separator + "core-site.xml"));
        configuration.addResource(new Path(confPath + File.separator + "hdfs-site.xml"));
        return configuration;
    }

    /**
     * 向HDFS指定目录创建一个文件
     *
     * @param fs       HDFS文件系统
     * @param dst      目标文件路径
     * @param contents 文件内容
     */
    public static void createFile(FileSystem fs, String dst, String contents) {
        try {
            Path path = new Path(dst);
            FSDataOutputStream fsDataOutputStream = fs.create(path);
            fsDataOutputStream.write(contents.getBytes());
            fsDataOutputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 上传本地文件至HDFS
     *
     * @param fs  HDFS文件系统
     * @param src 源文件路径
     * @param dst 目标文件路径
     */
    public static void uploadFile(FileSystem fs, String src, String dst) {
        try {
            Path srcPath = new Path(src); //原路径
            Path dstPath = new Path(dst); //目标路径
            //调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
            fs.copyFromLocalFile(false, srcPath, dstPath);
            //打印文件路径
            System.out.println("------------list files------------" + "\n");
            FileStatus[] fileStatus = fs.listStatus(dstPath);
            for (FileStatus file : fileStatus) {
                System.out.println(file.getPath());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 文件重命名
     *
     * @param fs
     * @param oldName
     * @param newName
     * @throws IOException
     */
    public static void rename(FileSystem fs, String oldName, String newName) {
        try {
            Path oldPath = new Path(oldName);
            Path newPath = new Path(newName);
            boolean isok = fs.rename(oldPath, newPath);
            if (isok) {
                System.out.println("rename ok!");
            } else {
                System.out.println("rename failure");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除文件
     *
     * @param fs
     * @param filePath
     * @throws IOException
     */
    public static void delete(FileSystem fs, String filePath) {
        try {
            Path path = new Path(filePath);
            boolean isok = fs.deleteOnExit(path);
            if (isok) {
                System.out.println("delete ok!");
            } else {
                System.out.println("delete failure");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 读取HDFS文件
     *
     * @param fs
     * @param filePath 文件路径
     */
    public static void readFile(FileSystem fs, String filePath) {
        try {
            Path srcPath = new Path(filePath);
            InputStream in = null;
            in = fs.open(srcPath);
            IOUtils.copyBytes(in, System.out, 4096, false);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
